package tedu.rocket;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.Test;

import java.nio.charset.StandardCharsets;

public class MyProducer {
    //发送消息测试
    @Test
    public void send() throws Exception {
        //1. 创建具有连接rocketmq能力的生产者对象
        DefaultMQProducer producer=new DefaultMQProducer();
        // 提供给生产者 nameserver地址
        producer.setNamesrvAddr("localhost:9876");
        // 每一个生产者 都属于一个分组,分组名称自定义
        producer.setProducerGroup("group01");
        // 启动连接 连接在初次创建的时候,可能因为环境原因建立比较缓慢.
        producer.start();
        //Thread.sleep(1000);// 如果连接建立失败,检查无果后可以添加等待时间
        //2. 组织一个消息对象,封装业务数据
        Message message=new Message();
        // 消息体 body 二进制 byte 如果传输的是对象数据 设计到序列化等
        String msg="我的第一条消息";
        byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
        message.setBody(bytes);
        message.setKeys("order:sn:001");
        message.setTags("上海");
        // 指定一个发送的明确的目标 rockentmq叫做主题
        message.setTopic("topic02");
        //3. 调用api方法将消息发送 同步发送
        SendResult send = producer.send(message);
        //4. 接收发送结果 发送的状态 rockemtq中一些路由信息
        // 打印发送结果
        SendStatus sendStatus = send.getSendStatus();
        System.out.println("发送状态:"+sendStatus);
        // 打印队列信息
        MessageQueue messageQueue = send.getMessageQueue();
        System.out.println("队列信息:"+messageQueue);
        //5. 关闭生产者连接
        producer.shutdown();
    }
}
